package com.framework.flink;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.table.expressions.E;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * 描述：
 * mysql的另一种读取方式
 * 基于流式数据
 * @author xianggj
 * @Date 2021/10/19 13:43
 **/
public class FlinkTestMysqlSource {
    public static void main(String[] args) throws Exception {
        // 调用程序
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 添加自定义数据源
        DataStreamSource<Tuple2<String, String>> data = env.addSource(new MySourceMysql());
        data.print().setParallelism(2);
        // 提交执行任务
        JobExecutionResult result = env.execute("MySourceMysql");
        System.out.println(result.getJobID());
    }

    public static class MySourceMysql extends RichSourceFunction<Tuple2<String, String>> {
        private PreparedStatement ps = null;
        private Connection connection = null;
        String driver = "com.mysql.jdbc.Driver";
        String url = "jdbc:mysql://192.168.1.37:3306/test";
        String username = "root";
        String password = "123456";

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            // 获取连接
            connection = getConn();
            //执行查询
            ps = connection.prepareStatement("select * from test;");
        }

        private Connection getConn() {
            try {
                Class.forName(driver);
                connection = DriverManager.getConnection(url, username, password);
            } catch (Exception e) {
                e.printStackTrace();
            }
            return connection;
        }

        @Override
        public void run(SourceContext<Tuple2<String, String>> ctx) throws Exception {
            ResultSet resultSet = ps.executeQuery();
            while (resultSet.next()) {
                String name = resultSet.getString("name");
                String id = resultSet.getString("id");
                Tuple2<String,String> tuple2 = new Tuple2<>();
                tuple2.setFields(id,name);
                ctx.collect(tuple2);//发送结果，结果是tuple2类型，2表示两个元素，可根据实际情况选择
            }
        }

        @Override
        public void cancel() {

        }

        @Override
        public void close() throws Exception {
            super.close();
            if(connection != null){
                connection.close();
            }
            if (ps != null){
                ps.close();
            }
        }
    }
}
